package com.wujian.love.redis;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import redis.clients.jedis.Jedis;

import java.lang.reflect.Type;
import java.util.Set;
import java.util.UUID;

/**
 * 延时队列实现    通过 Redis 的zset有序列表 来实现，讲消息序列化一个字符串作为zet 的 value ，
 * 这个消息的到期时间作为scorce,然后用多个线程 轮询zset 获取到期的任务进行处理。多个线程时保证可用性
 *
 *
 *
 * Redis作为消息队列为什么不能保证100%可靠
 * 使用lua 脚本实现
 */
public class RedisDelayingQueue<T> {

    static class TaskItem<T> {
        public String id;
        public T msg;
    }

    //fastjson 序列化对象中存在generic 类型时，需要使用TypeReference
    private Type TaskType = new TypeReference<TaskItem<T>>(){}.getType();


    private Jedis jedis;

    private String queueKey;

    public RedisDelayingQueue(Jedis jedis,String queueKey){
        this.jedis = jedis;
        this.queueKey = queueKey;
    }

    public void delay(T msg){
        TaskItem<T> task = new TaskItem<T>();
        task.id = UUID.randomUUID().toString();
        task.msg = msg;
        String s = JSON.toJSONString(task);
        //塞入延时队列，5s后再试
        jedis.zadd(queueKey,System.currentTimeMillis()+500,s);

    }

    public void loop(){
        while (!Thread.interrupted()){
            Set<String> values = jedis.zrangeByScore(queueKey,0,System.currentTimeMillis(),0,1);

            if(values.isEmpty()){
                try {
                    //歇会继续
                    Thread.sleep(500);
                }catch (InterruptedException e){
                    break;
                }
                continue;
            }

            String s = values.iterator().next();
            //抢到了
            if(jedis.zrem(queueKey,s)> 0){
                //反序列化
                TaskItem<T> task = JSON.parseObject(s,TaskType);
                this.handleMsg(task.msg);
            }
        }

    }

    public void handleMsg(T msg){
        System.out.println(msg);
    }




}